package generic.activemq;

import generic.Publisher;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnectionFactory;

public class ActiveMqRedundantPublisher implements Publisher {

	private Session session1;
	private Session session2;
	private MessageProducer messageProducer1;
	private MessageProducer messageProducer2;

	private String url1 = "tcp://ec2-54-242-185-60.compute-1.amazonaws.com:61616";
	private String url2 = "tcp://ec2-54-242-185-60.compute-1.amazonaws.com:61618";
//	private String url1 = "tcp://localhost:61616";
//	private String url2 = "tcp://localhost:61618";
	// private String url = "tcp://localhost:61616";
	// private String url =
	// "failover:(tcp://localhost:61616,tcp://localhost:61618)?randomize=false&backup=false&initialReconnectDelay=100&maxReconnectAttempts=1";
	// private String url =
	// "fanout:(static:(tcp://localhost:61616,tcp://localhost:61618))?minAckCount=1&initialReconnectDelay=1&maxReconnectDelay=1&maxReconnectAttempts=1";

	private StringBuilder builder;

	public ActiveMqRedundantPublisher(String url1, String url2) {
		super();
		this.url1 = url1;
		System.out.println("url1:" + url1);
		this.url2 = url2;
		System.out.println("url2:" + url2);

		builder = new StringBuilder();

		initialize1(url1);
		initialize2(url2);

		System.out
				.println("ActiveMqRedundantPublisher ready to publish messages...");
	}

	private void initialize1(String brokerUrl) {
		try {
			ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(
					brokerUrl);
			Connection connection;
			connection = factory.createConnection();
			// ((ActiveMQConnection)connection).setUseAsyncSend(false);
			// ((ActiveMQConnection)connection).setAlwaysSessionAsync(false);
			// ((ActiveMQConnection)connection).setAlwaysSyncSend(true);
			session1 = connection
					.createSession(false, Session.AUTO_ACKNOWLEDGE);
			Topic topic = session1.createTopic("topictest.messages");

			messageProducer1 = session1.createProducer(topic);
			messageProducer1.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

			connection.start();
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}

	private void initialize2(String brokerUrl) {
		try {
			ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(
					brokerUrl);
			Connection connection;
			connection = factory.createConnection();
			// ((ActiveMQConnection)connection).setUseAsyncSend(false);
			// ((ActiveMQConnection)connection).setAlwaysSessionAsync(false);
			// ((ActiveMQConnection)connection).setAlwaysSyncSend(true);
			session2 = connection
					.createSession(false, Session.AUTO_ACKNOWLEDGE);
			Topic topic = session2.createTopic("topictest.messages");

			messageProducer2 = session2.createProducer(topic);
			messageProducer2.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

			connection.start();
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}

	@Override
	public boolean send(String message) {
		boolean result1 = false;
		boolean result2 = false;
		TextMessage textMessage = null;
		try {
			if (messageProducer1 != null) {
				textMessage = session1.createTextMessage(message);
				messageProducer1.send(textMessage);
				result1 = true;
			}
		} catch (JMSException e) {
			messageProducer1 = null;
			//e.printStackTrace();
		}
		try {
			if (messageProducer2 != null) {
				textMessage = session2.createTextMessage(message);
				messageProducer2.send(textMessage);
				result2 = true;
			}
		} catch (JMSException e) {
			messageProducer2 = null;
			//e.printStackTrace();
		}
		return result1 || result2;
	}

}
